package metadata_manager

import (
	"gitee.com/zaper/kafka-service/service/metadata/model"
	"gitee.com/zaper/private-logger"
	"github.com/segmentio/kafka-go"
)

func (metadataManager *MetadataManager) GetAllPartitionsByTopicName(topicName string) []kafka.Partition {
	topic := metadataManager.GetTopicByName(topicName)
	if topic == nil {
		return nil
	}
	return topic.Partitions
}

func (metadataManager *MetadataManager) ChangePartitions(model *model.PartitionModel) {
	if !metadataManager.checkPartitionInfo(model) {
		return
	}
	// 构建 CreatePartitions 请求
	req := kafka.CreatePartitionsRequest{
		Topics: []kafka.TopicPartitionsConfig{{
			Name:  model.TopName,
			Count: model.NumPartitions,
		}},
	}
	// 执行分区修改
	resp, err := metadataManager.client.CreatePartitions(metadataManager.ctx, &req)
	if err != nil || resp.Errors[model.TopName] != nil {
		logger.Infof("修改分区失败: %v,resp err is %v", err, resp.Errors[model.TopName])
		return
	}
	logger.Infof("Current topic is partition count is %d", model.NumPartitions)
}

func (metadataManager *MetadataManager) checkPartitionInfo(model *model.PartitionModel) bool {
	meta, _ := metadataManager.client.Metadata(metadataManager.ctx, &kafka.MetadataRequest{
		Topics: []string{model.TopName},
	})
	if len(meta.Topics) == 0 {
		logger.Infof("Current topic %s is not exit", model.TopName)
		return false
	}
	logger.Infof("Current topic is partition count is %d", len(meta.Topics[0].Partitions))
	return true
}
